package com.shujia.flink.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTime {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 因为每一个并行度中单独计算水位线，如果数据量不大的时候会导致看不到效果
      *
      */

    env.setParallelism(1)


    //将flink程序的时间模式设置为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    /**
      * 事件事件：数据中自带的时间字段
      *
      */
    /*

001,1616142176000
001,1616142177000
001,1616142178000
001,1616142179000
001,1616142180000
001,1616142176000
001,1616142181000
001,1616142182000
001,1616142183000
001,1616142185000
     */


    val linesDS: DataStream[String] = env.socketTextStream("master", 7777)

    val events: DataStream[Event] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      Event(split(0), split(1).toLong)
    })

    //指定数据中哪一个列是事件时间
    //    val tsEvents: DataStream[Event] = events.assignAscendingTimestamps(_.ts)


    //指定时间字段和水位线前移的时间（允许数据最大延迟到达的时间）
    val tsEvents: DataStream[Event] = events.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
      override def extractTimestamp(element: Event): Long = element.ts
    })


    tsEvents
      .map(e => (e.id, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .sum(1)
      .print()


    env.execute()


  }

  case class Event(id: String, ts: Long)

}
